package com.atguigu.flink.state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by 黄凯 on 2023/6/20 0020 13:49
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * flink中计算可以分为有状态的计算和无状态的计算。
 * *      无状态计算： ETL
 * *                  每一条数据在计算时，相对独立，不会使用人和其他数据留下的信息。
 * *      有状态计算:  wordcount
 * *                  每一条数据需要和上一条数据留下的信息，继续运算。
 * *
 * *      状态： 在Task中保留的供后续数据计算使用的变量(数据)，称为状态。
 * *
 * *  -------------------------------------------------
 * *      分类:
 * *          RawState： 原始状态。客户自己定义和维护的状态。
 *                       RawState是java的类，比如 List
 * *                          维护： 定期备份状态，尝试在Task失败时，恢复状态。
 * *
 * *          ManagedState: 管理状态。使用Flink提供的状态功能定义的状态。Flink自动维护状态。
 */
public class Flink01_RawState {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        //设置重启策略。最多尝试3次重启，每次重试之间间隔2s
//        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,2000));

        //开启了checkpoint（备份），每间隔2s自动备份，程序就会无限次重启
        env.enableCheckpointing(2000);

        env.socketTextStream("127.0.0.1", 8888)
                .map(new MyMapFunction())
                .addSink(new SinkFunction<String>() {
                    @Override
                    public void invoke(String value) throws Exception {

                        if (value.contains("x")) {

                            System.out.println("抛出异常");

                            throw new RuntimeException("出异常了！！！");
                        }

//                        System.out.println("正常打印");
                        System.out.println(value);

                    }
                });

        env.execute();

    }

    /**
     * 功能：实现字符串的累积打印
     */
    public static class MyMapFunction implements MapFunction<String, String> {

        //用来累积字符串的集合，是一个RawState状态
        List<String> strs = new ArrayList<>();

        @Override
        public String map(String value) throws Exception {

            System.out.println("MyMapFunction.map");

            strs.add(value);
            return strs.toString();

        }
    }

}
